package com.ibm.cps.newmessage;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.messagequeue.AMQPClient;
import com.ibm.cps.messagequeue.MQTTMessageQueueClient;
import com.ibm.cps.newmessage.MetadataInputSchema.FieldType;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.IMessageClient;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class MQDataSourceMetadata extends AbstractDataSourceMetadata {

	private String topicName;
	private String hostname;
	private String protocol;
	private String username;
	private String password;
	private int port;
	private IMessageClient client;

	public MQDataSourceMetadata(String input) throws CPSException {
		this(JSONUtility.fromObjectString(input));
	}

	/**
	 * @param input
	 * @throws CPSException
	 */
	public MQDataSourceMetadata(ObjectNode object) throws CPSException {
		super(object);
		setTimeSeriesSchema(schemaString);
		this.topicName = JSONUtility.GetString(MessageFields.TOPICNAME, object);
		this.hostname = JSONUtility.GetString(MessageFields.HOSTNAME, object);
		this.port = JSONUtility.GetInteger(MessageFields.PORT, object, 0);
		this.protocol = JSONUtility.GetString(MessageFields.PROTOCOL_TYPE, object);
		this.username = JSONUtility.GetString(MessageFields.USERNAME, object);
		this.password = JSONUtility.GetString(MessageFields.PASSWORD, object);
		setDataSourceConsumerClient();
	}

	public String toString() {
		return input;
	}

	public ITimeSeriesSchema getTimeSeriesSchema() throws CPSException {
		return setTimeSeriesSchema(schemaString);
	}

	private ITimeSeriesSchema setTimeSeriesSchema(String schemaString) throws CPSException {
		JsonNode jsonSchema = JSONUtility.fromString(schemaString);
		return new TimeSeriesSchema(tenantid, schemaid, (ArrayNode) jsonSchema, -1);
	}

	@Override
	public MetadataInputSchema getInputMetadataSchema() {
		String[] commonFields = { MessageFields.TYPE, MessageFields.TENANTID, MessageFields.NAME,
				MessageFields.TOPICNAME, MessageFields.SCHEMA_FORMAT_TYPE };

		String[] serverFields = { MessageFields.HOSTNAME, MessageFields.PROTOCOL_TYPE, MessageFields.USERNAME,
				MessageFields.PASSWORD };

		MetadataInputSchema inputSchema = new MetadataInputSchema();
		inputSchema.addField(FieldType.String, commonFields);
		inputSchema.addField(FieldType.String, serverFields);
		inputSchema.addField(FieldType.Array, MessageFields.SCHEMA);
		inputSchema.addField(FieldType.Integer, MessageFields.PORT);
		return inputSchema;
	}

	@Override
	public IMessageClient getDataSourceConsumerClient() throws CPSException {
		return client;
	}

	private void setDataSourceConsumerClient() throws CPSException {
		if (protocol == null) {
			throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "Protocol type should not be null.");
		}
		if (protocol.equals("amqp")) {
			client = new AMQPClient(hostname, port, username, password);
		} else if (protocol.equals("mqtt")) {
			client = new MQTTMessageQueueClient(hostname, port, username, password);
		} else {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
					"Unknown protocal type " + protocol + " for data source type " + type);
		}
	}

	@Override
	public String getSubscribeTarget() {
		return topicName;
	}
}
